package com.atguigu.chapter02;


import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
public class BatchWordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 2. 从文件读取数据 按行读取(存储的元素就是每行的文本)
        DataSource<String> lineDS = env.readTextFile("input/words.txt");

        lineDS.flatMap( (String line,Collector<Tuple2<String,Long>> out) ->{
            String[] words = line.split(" ");
            for(String word: words)
                out.collect(Tuple2.of(word,1L));
        })
//                .returns(new TypeHint<Tuple2<String, Long>>() {})
                .returns(Types.TUPLE(Types.STRING,Types.LONG)).groupBy(0).sum(1).print();

    }
}